package com.datamining.online

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

/**
  * Created by Administrator on 2017/5/6.
  */
/**
  * spark-test
  * HiveTest
  *
  * @author Administrator kevin
  * @create 2017-05-06 21:15
  */


object HiveTestBySecond {

  case class TimeQuantity(time: Long, quantity: Integer);

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf();
    sparkConf.setMaster("local[4]"); // 本地模式
    sparkConf.setAppName("my_test");

    //    val sc = new SparkContext(sparkConf)
    // 创建sparkSession
    val sparkSession = SparkSession.builder().appName("").config(conf = sparkConf).enableHiveSupport().getOrCreate();

    //    val hiveContext = new HiveContext(sc)
    //
    //    hiveContext.sql("use yamibuy_master")
    //    val resultDF = hiveContext.sql("SELECT order_id, order_sn, order_status, pay_status, user_id FROM yamibuy_master.xysc_order_info LIMIT 10")

    import sparkSession.sql

        val sqlDataFrameOrderInfo = sql("SELECT order_id, order_sn, order_status, pay_status, user_id, add_time, pay_time FROM yamibuy_master.xysc_order_info WHERE pay_status > 0")
    //    val sqlDataFrameOrderInfo = sql("SELECT order_id, order_sn, order_status, pay_status, user_id, from_unixtime(add_time, 'yyyy-MM-dd') AS add_time, from_unixtime(pay_time, 'yyyy-MM-dd') AS pay_time FROM yamibuy_master.xysc_order_info WHERE pay_status > 0")

    val sqlDataFrameOrderGoods = sql("SELECT order_id, goods_id, goods_number FROM yamibuy_master.xysc_order_goods")

    val order_item_join = sqlDataFrameOrderInfo.join(sqlDataFrameOrderGoods, "order_id")

    val user_item_add_time_desc = order_item_join.orderBy(order_item_join("add_time").desc)
    val tmpVeiw = user_item_add_time_desc.createOrReplaceTempView("user_item_add_time_desc")

    val sqlDataFrameGoodsGoodsNumber = sql("SELECT goods_id, add_time, SUM(goods_number) AS goods_number FROM user_item_add_time_desc WHERE goods_id > 0 GROUP BY goods_id, add_time ORDER BY goods_id ASC,add_time DESC")
//    sqlDataFrameGoodsGoodsNumber.show(10)
    val sqlDataFrameGoodsGoodsNumberRdd = sqlDataFrameGoodsGoodsNumber.rdd

    val ttt_rdd = sqlDataFrameGoodsGoodsNumberRdd.map(x => (x(0), (x(1), x(2), 0.toDouble)))

    // add_time倒序
    val ttt_rdd_sort = ttt_rdd.groupByKey().sortBy(x => x._2.map(y => Integer.valueOf(y._1.toString)), false)

    // 每个商品保留最近100条记录
    val ttt_rdd_sort_sub = ttt_rdd_sort.map(x => (x._1, x._2.splitAt(100)._1))

    val item_repurchasing = ttt_rdd_sort_sub.map(x => {
      val item_time_quantity_array = x._2.toArray
      val length = item_time_quantity_array.size
      for (i <- 0 to length - 2) {

        val item_1 = item_time_quantity_array.apply(i)
        val item_2 = item_time_quantity_array.apply(i + 1)

        val time_1 = item_1._1
        val time_2 = item_2._1
        val quantity = item_2._2

        // 按秒
        val Tn = (time_1.toString.toDouble - time_2.toString.toDouble) / quantity.toString.toDouble
//        // 按天
//        val Tn = compareDate(time_1.toString, time_2.toString).toDouble / quantity.toString.toDouble

        item_time_quantity_array.update(i, (item_1._1, item_1._2, Tn))
      }
      (x._1, item_time_quantity_array)
    })

    item_repurchasing.take(10)
    item_repurchasing.count()


  }
}
